# encoding: utf-8

import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark import SparkFiles

df1 = spark.sql("select * from idealsh.add_all_month_202107 where offer_flag_5g3 =1 limit 1000000")
df2 = spark.sql("select * from idealsh.add_all_month_202108 where offer_flag_5g3 =1 limit 1000000")
df3 = spark.sql("select * from idealsh.add_all_month_202109 where offer_flag_5g3 =1 limit 1000000")
df4 = spark.sql("select * from idealsh.add_all_month_202110 where offer_flag_5g3 =1 limit 1000000")

pdf7 = df1.toPandas()
pdf8 = df2.toPandas()
pdf9 = df3.toPandas()
pdf10 = df4.toPandas()

pdf7['month']='201107'
pdf8['month']='201108'
pdf9['month']='201109'
pdf10['month']='201110'
pdf1 = pd.concat([pdf7, pdf8])
pdf2 = pd.concat([pdf9, pdf10])
pdf3 = pd.concat([pdf1, pdf2])
print(pdf3.head(10))
print(pdf3.shape)

pdf3.month.value_counts()

for i in ['202107','202108','202109','202110']:
    pdf3.loc[pdf3['month']==i,['fair_amount_3avg','arpu_average_3']].head(10)

# str字段处理
strcolumns=['business_id',
'lte_card_type',
'channel_type',
'net_flag',
'tianyi_offer_name',
'network_standard',
'brand_name',
'cust_income_level']

# pdf3[strcolumns] = pdf3[strcolumns].fillna(method='ffill')
pdf3[strcolumns] = pdf3[strcolumns].fillna('')


for i in ['201107','201108','201109','201110']:
    pdf3[pdf3['month']==i][strcolumns].info()

pdf3['tianyi_offer_name'] = pdf3['tianyi_offer_name'].fillna('')
pdf3['tianyi_offer_name_new'] = pdf3['tianyi_offer_name'].apply(lambda x: x.replace('（','('))

for i in ['202107','202108','202109','202110']:
    pdf3[pdf3['month']==i]['tianyi_offer_name_new'].value_counts()[:10]

for i in ['201107','201108','201109','201110']:
    pdf3[pdf3['month']==i]['business_id'].value_counts()[:10]

for i in ['201107','201108','201109','201110']:
    pdf3[pdf3['month']==i]['channel_type'].value_counts()[:10]

pdf3['brand_name'] = pdf3['brand_name'].apply(lambda x: x.lower())
pdf3.loc[(pdf3['brand_name'] == 'apple'), 'brand_name'] = '苹果'
pdf3.loc[(pdf3['brand_name'] == '维沃'), 'brand_name'] = 'vivo'
pdf3.loc[(pdf3['brand_name'] == '欧珀'), 'brand_name'] = 'oppo'

for i in ['201107','201108','201109','201110']:
    pdf3[pdf3['month']==i]['brand_name'].value_counts()[:15]